package org.jeecg.modules.middleware.rabbitmq.delay;


import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import org.apache.commons.lang.StringUtils;
import org.jeecg.common.exception.JeecgBootException;
import org.jeecg.modules.middleware.rabbitmq.delay.tl.TlDelayHandler;
import org.jeecg.modules.middleware.rabbitmq.delay.tl.TlDelayMessageType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 延时消息消费者
 */
@Component
public class DelayMessageConsumer {
    private ConcurrentHashMap<String, IDelayMqHandle> hanlers= new ConcurrentHashMap <>();

    @Autowired
    public DelayMessageConsumer(List<IDelayMqHandle> handleList) {
        for (IDelayMqHandle handle : handleList) {
            String handlerName = handle.getHandlerName();
            if(StringUtils.isEmpty(handlerName)){
                throw new JeecgBootException("delay handler error");
            }
            if(hanlers.containsKey(handlerName)){
                throw new JeecgBootException("delay handler name 重复");
            }
            this.hanlers.put(handlerName, handle);
        }
    }

    @Autowired
    private TlDelayHandler tlDelayHandler;

    //通用延时消息处理
    @RabbitListener(queues = DelayMqConst.GEN_DELAYED_QUEUE_NAME)
    public void handleGenDelayMessage(String msg , Message message, Channel channel) throws Exception {
        try {
            JSONObject jo = JSONObject.parseObject(msg);
            String handlerName = jo.getString("handlerName");
            IDelayMqHandle handler = hanlers.get(handlerName);
            if (handler != null) {
                String msgType = jo.getString("msgType");
                String msgContent = jo.getString("msg");
                handler.handle(msgType, msgContent);
            }
        } catch (Exception e) {
            e.printStackTrace();
            /**
             * basicRecover方法是进行补发操作，
             * 其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer(集群)接收到，
             * 设置为false是只补发给当前的consumer
             */
//            channel.basicRecover(false);
        } finally {
            //消费消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    //通联延时消息处理
    @RabbitListener(queues = DelayMqConst.TL_DELAYED_QUEUE_NAME)
    public void handleTLDelayMessage(String msg , Message message, Channel channel) throws Exception {
        try {
            JSONObject jo = JSONObject.parseObject(msg);
            TlDelayMessageType msgType = jo.getObject("msgType", TlDelayMessageType.class);
            String msgContent = jo.getString("msg");
            tlDelayHandler.handle(msgType, msgContent);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //消费消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}
